package cn.whitecao.agent.core.repoter;

import cn.whitecao.agent.core.config.ConfigContext;
import cn.whitecao.agent.core.model.Span;
import cn.whitecao.agent.core.repoter.impl.OkhttpReporter;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.IntStream;

/**
 * 上报消息处理器
 *  使用生产者消费模型把数据
 *  定期提交数据
 *
 * @author : BashfulHuang
 **/
@Slf4j
public final class ReporterProcessor {
    /**
     * 缓存需要上报的采集数据
     */
    private static BlockingQueue<Span> reportCacheQueue;

    /**
     * 初始化的状态
     */
    volatile private static Boolean initStatus=false;
    /**
     *  初始化
     * 1.读取配置文件
     * 2.初始化队列
     * 3.初始化上报实现
     * 4.启动线程开始上报
     */
    public static synchronized void init(){
        if (initStatus){
            return;
        }
        log.info("初始化上报数据处理器");
        reportCacheQueue=new LinkedBlockingQueue<>(ConfigContext.getInt(ReporterConfig.REPORTER_QUEUE_SIZE,2000));
        String reportTarget = ConfigContext.getStr(ReporterConfig.REPORTER_TARGET, null);
        Integer threadNum = ConfigContext.getInt(ReporterConfig.REPORTER_SUBMIT_THREAD, 3);
        //每次提交的个数
        Integer perSubmitNum = ConfigContext.getInt(ReporterConfig.REPORTER_SUBMIT_NUM, 100);
        Integer submitInterval = ConfigContext.getInt(ReporterConfig.REPORTER_SUBMIT_INTERVAL, 2000);

        OkhttpReporter reporter = new OkhttpReporter();
        reporter.init(reportTarget);
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(threadNum,r->{
            //设置成守护的线程,如果不设置的话 main方法进来的被监控程序退出后 进程不会结束
            Thread thread = new Thread(r);
            thread.setDaemon(true);
            return thread;
        });
        IntStream.range(0,threadNum).forEach(i-> scheduledThreadPool.scheduleAtFixedRate(()->{
            try {
                List<Span> toSubmit=new ArrayList<>();
                reportCacheQueue.drainTo(toSubmit,perSubmitNum);
                if (toSubmit.size()==0){
                    return;
                }
                reporter.report(toSubmit);
                log.info("上报" + toSubmit.size()+"条数据");
            } catch (Exception e) {
                log.info("提交失败");
            }
        },0,submitInterval,TimeUnit.MILLISECONDS));
        initStatus=true;
    }

    /**
     * 上报数据
     * @param span collectDataEntity
     */
    public static void report(Span span){
        reportCacheQueue.offer(span);

    }



    private static class ReporterConfig{
        /**
         * 提交目标
         */
        static final String REPORTER_TARGET="reporter.target";
        /**
         * 缓存队列的大小
         */
        static final String REPORTER_QUEUE_SIZE="reporter.queueSize";

        /**
         * 每次提交的数量
         */
        static final String REPORTER_SUBMIT_NUM="reporter.submit.perNum";

        /**
         * 每次提交的间隔
         */
        static final String REPORTER_SUBMIT_INTERVAL="reporter.submit.interval";

        /**
         * 提交的线程数
         */
        static final String REPORTER_SUBMIT_THREAD="reporter.submit.thread";
    }

}
